package cn.doitedu.rtdw.dashboard;

import org.apache.flink.api.common.RuntimeExecutionMode;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.TableConfig;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

public class Job07_订单日清日结看板 {

    public static void main(String[] args) {

        // 创建环境
        // 创建编程环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.enableCheckpointing(5000, CheckpointingMode.EXACTLY_ONCE);
        env.getCheckpointConfig().setCheckpointStorage("file:///d:/ckpt");
        env.setParallelism(1);
        env.setRuntimeMode(RuntimeExecutionMode.STREAMING);

        StreamTableEnvironment tenv = StreamTableEnvironment.create(env);
        TableConfig config = tenv.getConfig();
        // 开启窗口的提前触发器（trigger）
        config.getConfiguration().setBoolean("table.exec.emit.early-fire.enabled", true);
        config.getConfiguration().setString("table.exec.emit.early-fire.delay", "1000 ms");


        // 创建业务库中的oms_order表的cdc连接器表
        tenv.executeSql(
                        " create table oms_order_mysql_cdc(                  \n "+
                        "   id           bigint,                             \n "+ // 订单id
                        "   status         int,                             \n "+ // 订单状态
                        " 	total_amount decimal(10,2),                      \n "+ // 订单金额
                        " 	pay_amount   decimal(10,2),                      \n "+ // 订单应付（实付）额
                        " 	create_time  timestamp(3),                       \n "+  // 订单创建时间
                        " 	payment_time  timestamp(3),                      \n "+  // 订单支付时间
                        " 	delivery_time  timestamp(3),                     \n "+  // 订单发货时间
                        " 	receive_time  timestamp(3),                      \n "+  // 订单完成时间
                        " 	modify_time  timestamp(3),                      \n "+
                        " 	rt as modify_time,                              \n "+
                        " 	watermark for rt as rt - interval '0' second,   \n "+
                        "   primary key (id) not enforced                   \n "+
                        " )WITH (                                           \n "+
                        "     'connector' = 'mysql-cdc',                    \n "+
                        "     'hostname' = 'doitedu',                       \n "+
                        "     'port' = '3306',                              \n "+
                        "     'username' = 'root',                          \n "+
                        "     'password' = 'root',                          \n "+
                        "     'database-name' = 'realtimedw',               \n "+
                        "     'table-name' = 'oms_order'                    \n "+
                        " )                                                 \n "
        );

        // 核心逻辑：累计窗口
        tenv.executeSql(
                        " SELECT	                                                                                            "+
                        "  tumble_start(rt,interval '24' hour) 	 as window_start                                                "+
                        " ,count(1) filter(where date_format(create_time,'yyyy-MM-dd') = current_date)                           "+
                        " ,sum(total_amount)  filter(where date_format(create_time,'yyyy-MM-dd') = current_date)                "+
                        " ,sum(pay_amount)  filter(where date_format(create_time,'yyyy-MM-dd') = current_date)                  "+
                        " ,count(1) filter(where date_format(create_time,'yyyy-MM-dd') = current_date and status=0 )            "+
                        " ,sum(total_amount) filter(where date_format(create_time,'yyyy-MM-dd') = current_date and status=0 )   "+
                        " ,count(1) filter(where date_format(payment_time,'yyyy-MM-dd') = current_date)                         "+
                        " ,sum(total_amount) filter(where date_format(payment_time,'yyyy-MM-dd') = current_date)                "+
                        " ,count(1) filter(where date_format(delivery_time,'yyyy-MM-dd') = current_date)                        "+
                        " ,sum(total_amount) filter(where date_format(delivery_time,'yyyy-MM-dd') = current_date)               "+
                        " ,count(1) filter(where date_format(receive_time,'yyyy-MM-dd') = current_date)                         "+
                        " ,sum(total_amount) filter(where date_format(receive_time,'yyyy-MM-dd') = current_date)                "+
                        " from oms_order_mysql_cdc                                                                              "+
                        " group by                                                                                              "+
                        "     tumble(rt,interval '24' hour)                                                                     "
        ).print();




    }

}
